package ItemBased;

import java.io.IOException;
import java.util.Iterator;
import java.util.Vector;

import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;


public class TopKSimilarity {
	
	public static class TopKSimilaritysMap 
	extends MapReduceBase
    implements Mapper<LongPairWritable, DoubleWritable, LongWritable, LongDoublePairWritable>{
		
	      
	    public void map(LongPairWritable key, DoubleWritable value, OutputCollector<LongWritable, LongDoublePairWritable> output, Reporter reporter
	                    ) throws IOException {
	    	LongWritable firstID = key.getFirst();
	    	LongWritable secondID = key.getSecond();
	    	
	    	output.collect(firstID, new LongDoublePairWritable(secondID.get(),value.get()));
	    	output.collect(secondID, new LongDoublePairWritable(firstID.get(),value.get()));
	    	
	    }		
	}

	public static class TopKSimilarityReduce
	extends MapReduceBase
    implements Reducer<LongWritable,LongDoublePairWritable,LongWritable,LongDoublePairArrayWritable> {
		
		public void reduce(LongWritable key, Iterator<LongDoublePairWritable> values, 
				OutputCollector<LongWritable, LongDoublePairArrayWritable> output, Reporter reporter
                ) throws IOException {
				
		}
	}
	
	

}
